package com.ruyuan.engine.core;
import com.ruyuan.engine.datagen.RuleComplexParamDataGen;
import com.ruyuan.engine.pojo.ClientLog;
import com.ruyuan.engine.pojo.RuleComplexParam;
import com.ruyuan.engine.pojo.RuleResult;
import com.ruyuan.engine.pojo.TriggerTimer;
import com.ruyuan.engine.router.Router;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.List;

/**
 * 基于定时器
 * */
public class RuleEngineProcessDynamicByTimer extends KeyedProcessFunction<String, ClientLog, RuleResult> {
    protected static final Logger log = LoggerFactory.getLogger(RuleEngineProcessDynamicByTimer.class);
    ListState<ClientLog> clientLogListState;
    RuleComplexParam ruleComplexParam;
    Router router;
    ListState<Tuple2<RuleComplexParam,Long>> ruleTimerState;


    @Override
    public void open(Configuration parameters) throws Exception {
        ListStateDescriptor<ClientLog> descriptor = new ListStateDescriptor<ClientLog>("clientLogListState",ClientLog.class);
        StateTtlConfig stateTtlConfig = StateTtlConfig.newBuilder(Time.hours(6)).updateTtlOnCreateAndWrite().build();
        descriptor.enableTimeToLive(stateTtlConfig);
        clientLogListState = getRuntimeContext().getListState(descriptor);
        ruleComplexParam = RuleComplexParamDataGen.mockData();
        router = new Router(clientLogListState);
        ruleTimerState = getRuntimeContext().getListState(new ListStateDescriptor<Tuple2<RuleComplexParam, Long>>("ruleTimerState", TypeInformation.of(new TypeHint<Tuple2<RuleComplexParam, Long>>() {
        })));
    }

    @Override
    public void processElement(ClientLog clientLog, Context ctx, Collector<RuleResult> out) throws Exception {
        clientLogListState.add(clientLog);
        if(router.isMatchRule(clientLog,ruleComplexParam)) {
            //基于ruleComplexParam 判断是否需要后续的定时处理的逻辑
            if (ruleComplexParam.isRegisterTimer()) {
                List<TriggerTimer> triggerTimers = ruleComplexParam.getTriggerTimers();
                for (TriggerTimer triggerTimer:triggerTimers) {
                    long detailTime = Long.parseLong(clientLog.getDetailTime());
//                    long timeLimit = ruleComplexParam.getTriggerTimers().get(0).getTimeLimit();
                    //定时器的时间
                    long timeLimit = triggerTimer.getTimeLimit();
                    //注册定时器
                    ctx.timerService().registerEventTimeTimer(detailTime+timeLimit);
                    //保存定时器注册时间+triggerTimer
                    ruleTimerState.add(Tuple2.of(ruleComplexParam,detailTime+timeLimit));
                }

            }
        }else {
            out.collect(new RuleResult(ruleComplexParam.getRuleId(),clientLog.getEventId(),System.currentTimeMillis()));
        }

    }

    /**
     * 定时规则满足逻辑结果处理
     * */
    @Override
    public void onTimer(long timestamp, OnTimerContext ctx, Collector<RuleResult> out) throws Exception {
        Iterable<Tuple2<RuleComplexParam, Long>> tuple2s = ruleTimerState.get();
        for(Tuple2<RuleComplexParam,Long> tuple:tuple2s) {
            if (tuple.f1 == timestamp) {
                TriggerTimer triggerTimer = tuple.f0.getTriggerTimers().get(0);
                /**
                 * 查询操作：
                 *      1. 查询State
                 *      2. HBase
                 *      3. clickhouse
                 * */
                if (router.isMatchRuleByTimer(ctx.getCurrentKey(),triggerTimer,timestamp - triggerTimer.getTimeLimit(),timestamp)) {
                    out.collect(new RuleResult(ruleComplexParam.getRuleId(),ctx.getCurrentKey(),System.currentTimeMillis()));
                }

            }
        }
    }
}
